package Market;

import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import Market.exchange.MarketData;
import Market.exchange.MarketDataSer;
import Market.exchange.ReadFromDirextory;;

public class exchangeDataProdcue {

    public static void main(String[] argv) throws ParseException {
        int accelerated = 10;
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, MarketDataSer.class);
        KafkaProducer<String, MarketData> producer = new KafkaProducer<>(props);

        ReadFromDirextory marketData = new ReadFromDirextory();
        marketData.readFromDirectory("data/quotedata_csv");
        // marketData.shuffee();
        long startTime = new Date().getTime();// ms
        // long baseTime =
        // sdf.parse(marketData.exchangeMarketData.get(0).getRecordTime()).getTime();
        long baseTime = marketData.exchangeMarketData.get(0).get(0).getRecordTime();
        for (int i = 0; i < marketData.getMaxSize(); i++) {
            for (ArrayList<MarketData> marketDataList : marketData.exchangeMarketData) {
                if (i >= marketDataList.size())
                    continue;
                MarketData oneMarketData = marketDataList.get(i);
                if(oneMarketData.getLast()==0)continue;

                long nowTime = new Date().getTime();
                long curTime = oneMarketData.getRecordTime();
                long diff = nowTime - startTime - (curTime - baseTime);
                if (diff < 0) {
                    try {
                        System.out.printf("sleep %d ms\n", 0 - diff / accelerated);
                        Thread.sleep(0 - diff / accelerated);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }

                producer.send(new ProducerRecord<>("stream-in", oneMarketData.getKey(), oneMarketData));

            }
        }
        producer.close();
    }
}
